package com.linys.scala.qf.day06_spark

import org.apache.spark.{SparkConf, SparkContext}

/**
  * action类算子
  */
object ExeciseAggregate_05 {



  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("ExeciseAggregate05").setMaster("local[*]")
    val sc = new SparkContext(conf)

    val rdd1 = sc.parallelize(List(2,1,3,6,5),2)

    println(rdd1.reduce(_+_))
    println(rdd1.count)
    println(rdd1.top(4).toBuffer)
    println(rdd1.take(3).toBuffer)
    println(rdd1.takeOrdered(2).toBuffer)
    println(rdd1.first)

    println("----------------------")
    val rdd2 = sc.parallelize(List(("a", 1), ("b", 2), ("b", 2), ("c", 2), ("c", 1)))
    println(rdd2.countByKey)
    println(rdd2.countByValue)

    println("----------------------")
    /**
      * foreach和foreachPartition的区别：
      * 首先他们都没有返回值
      * foreach是将func作用于每一个元素上，而foreachPartitions是将func作用于每个分区上，
      * 应用场景：一般都是用在将结果输出的场景，
      * 如果结果数据量很少，可以用foreach进行存储，
      * 如果数据量比较大，会拿很多的连接进行存储，可能数据库会直接宕机，可以用foreachPartiton，用一个分区对应一个连接
      */
    val rdd3 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 4)
    //rdd3.foreach(x => println(x))
    rdd3.foreachPartition(x => println(x.reduce(_+_)))

    sc.stop

  }

}
